package DianShang_2024.ds_03.tzgc


import org.apache.spark.sql.{Dataset, Row, SparkSession, functions}
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{col, dense_rank, lit, split}
import org.apache.spark.sql.types.DoubleType

import java.util.Properties

object tzgc02 {
  def main(args: Array[String]): Unit = {
    /*
        2、根据第1小题的结果，对其进行聚合，其中对sku_id进行one-hot转换，将其转换为以下格式矩阵：第一列为用户id，其余列
        名为商品id，按照用户id进行升序排序，展示矩阵第一行前5列数据，将结果截图粘贴至客户端桌面【Release\任务C提交结
        果.docx】中对应的任务序号下。
    字段	类型	中文含义	备注
    user_id	double	客户key
    sku_id0	double	用户是否购买过商品1	若用户购买过该商品，则值为1，否则为0
    sku_id1	double	用户是否购买过商品2	若用户购买过该商品，则值为1，否则为0
    sku_id2	double	用户是否购买过商品3	若用户购买过该商品，则值为1，否则为0
    .....
    结果格式如下：
    ---------------第一行前5列结果展示为---------------
    0.0,1.0,0.0,0.0,0.0
     */
    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("第三套特征工程第二题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    //  导入隐式转化避免不必要的报错
    import spark.implicits._

    //  首先拿到第一题的结果
    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")


val order_info=spark.read
  .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_info",connect)


  val order_detail=spark.read
    .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_detail",connect)

    //  下面多了一条操作把sku_id的值换成了sku_id+索引
    val r1:Dataset[Row]=order_info.join(order_detail,order_detail("order_id")===order_info("id"))
      .select("user_id","sku_id")
      .withColumn(
        "user_id",
        dense_rank().over(Window.orderBy("user_id"))-1
      )
      .withColumn(
        "sku_id",
        dense_rank().over(Window.orderBy("sku_id"))-1
      )
      .withColumn(
        "sku_id",
       functions.concat(lit("sku_id"),col("sku_id"))
      )
      .orderBy("user_id","sku_id")

//    r1.show(false)

    //  拿到所有的商品id
    //  1.因为是要拿到所有种类的商品id,所以需要去重
    //   2.split(col("sku_id"),"id")(1).cast("int"):首先对sku_id字段进行分割，然后返回索引1，也就是得到了数字，然后
    //  要将数字转换为int才可以排序，否则字符串不可以排序
    val sku_ids:Array[String]=r1.select("sku_id")
      .distinct()
      .orderBy(
        split(col("sku_id"),"id")(1).cast("int")
      )
      .map(
       x => x(0).toString
      )
      .collect()

    println("所有的sku_id")
    println(sku_ids.mkString(","))


//  1.pivot("sku_id", sku_ids):使用 pivot 操作将 sku_id 列的值变成列名，并根据 sku_ids 中的值填充数据
//  2.agg(lit(1.0)):在进行聚合时，对于每个用户和每个 SKU，填入一个常量值 1.0。这表示在每个 SKU 列下，如果用户存在对
//  应sku的记录，则该列的值为1.0,否则为null
//  3.na.fill(0.0)将所有的null替换为0.0

    val result=r1.groupBy("user_id")
      .pivot("sku_id",sku_ids)
      .agg(lit(1.0))
      .na.fill(0.0)
      .withColumn(
        "user_id",
        col("user_id").cast(DoubleType)
      )
      .orderBy("user_id")
      .show()





    spark.close()
  }

}
